package yuekao2.machine;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.SplitBatchOp;
import com.alibaba.alink.operator.batch.evaluation.EvalClusterBatchOp;
import com.alibaba.alink.operator.batch.evaluation.EvalMultiClassBatchOp;
import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.alibaba.alink.operator.common.evaluation.ClusterMetrics;
import com.alibaba.alink.operator.common.evaluation.MultiClassMetrics;
import com.alibaba.alink.operator.common.evaluation.TuningMultiClassMetric;
import com.alibaba.alink.pipeline.PipelineModel;
import com.alibaba.alink.pipeline.classification.KnnClassifier;
import com.alibaba.alink.pipeline.classification.RandomForestClassifier;
import com.alibaba.alink.pipeline.clustering.KMeans;
import com.alibaba.alink.pipeline.dataproc.vector.VectorAssembler;
import com.alibaba.alink.pipeline.tuning.GridSearchCV;
import com.alibaba.alink.pipeline.tuning.MultiClassClassificationTuningEvaluator;
import com.alibaba.alink.pipeline.tuning.ParamGrid;

public class WineMachine {
    public static void main(String[] args) throws Exception {
        //创建Flink程序，调用Alink API，设置批处理操作的并行度为1，加载葡萄酒数据集，提取特征features和目标值label，输出控制台；（3分）
        String filePath = "data/yk2/wine.data";
        String schema
                //1,    14.23,  1.71,   2.43,   15.6,   127,    2.8,    3.06,   .28,    2.29,   5.64,   1.04,   3.92,   1065
                = "label int,f0 double, f1 double, f2 double, f3 double, f4 double, f5 double, f6 double, f7 double, f8 double, f9 double" +
                ", f10 double, f11 double, f12 double";
        CsvSourceBatchOp csvSource = new CsvSourceBatchOp()
                .setFilePath(filePath)
                .setSchemaStr(schema)
                .setFieldDelimiter(",")
                .setIgnoreFirstLine(true)
                .setSkipBlankLine(true)
                .setLenient(true);

//        csvSource.print();
        BatchOperator.setParallelism(1);
        //提取特征 features 和目标值 label
        String[] features=new String[]{"f0","f1","f2","f3","f4","f5","f6","f7","f8","f9","f10","f11","f12"};
        String label="label";
        //（2）、调用Alink 特征工程API，分割数据集为3份，分别为：训练数据集trainData、验证数据集validateData 和 测试数据集testData，统计数据条目数，输出控制台；（3分）
        BatchOperator<?> spliter = new SplitBatchOp().setFraction(0.6);
        //0.6
        BatchOperator<?> trainData = spliter.linkFrom(csvSource);
        BatchOperator<?> sideOutput = spliter.getSideOutput(0);

        BatchOperator<?> spliter1 = new SplitBatchOp().setFraction(0.5);
        //0.2
        BatchOperator<?> validateData = spliter1.linkFrom(sideOutput);
        //0.2
        BatchOperator<?> testData = spliter1.getSideOutput(0);

        System.out.println("训练数据集trainData:"+trainData.count());
        System.out.println("验证数据集validateData:"+validateData.count());
        System.out.println("测试数据集testData:"+testData.count());

        //（3）、构建分类模型：对训练数据集分别使用KNN紧邻算法和随机森林RF算法，构建训练不同分类模型，合理设置参数值；（5分）
        //KNN紧邻算法
        KnnClassifier knn = new KnnClassifier()
                .setPredictionCol("pred")
                .setLabelCol("label")
                .setFeatureCols(features)
                .setPredictionDetailCol("pred_detail")
                .setDistanceType("EUCLIDEAN")
                .setK(3);

        BatchOperator<?> knn_transform = knn.fit(trainData).transform(validateData);
        //随机森林RF算法
        RandomForestClassifier rfc = new RandomForestClassifier()
                .setPredictionDetailCol("pred_detail")
                .setPredictionCol("pred")
                .setLabelCol("label")
                .setFeatureCols(features)
                .setMaxBins(128)
                .setMinInfoGain(0.0);

        BatchOperator<?> rfc_transform = rfc.fit(trainData).transform(validateData);

        //（4）、构建分类模型：调用Alink中批组件分类评估API，使用验证数据集预测评估上述构建2个不同分类模型性能，评估指标：准确率Accuracy和召回率Recall；（5分）
        EvalMultiClassBatchOp metrics = new EvalMultiClassBatchOp()
                .setLabelCol("label")
                .setPredictionDetailCol("pred_detail");
        //knn
        MultiClassMetrics knn_multiClassMetrics = metrics.linkFrom(knn_transform).collectMetrics();
        System.out.println("knn accuracy:" + knn_multiClassMetrics.getAccuracy());
        System.out.println("knn recall:" + knn_multiClassMetrics.getMacroRecall());
        //rfc
        MultiClassMetrics rfc_multiClassMetrics = metrics.linkFrom(rfc_transform).collectMetrics();
        System.out.println("rfc accuracy:" + rfc_multiClassMetrics.getAccuracy());
        System.out.println("rfc recall:" + rfc_multiClassMetrics.getMacroRecall());

        //（5）、构建分类模型：对KNN算法和RandomForest算法超参数设置不同的值（每个算法至少2个超参数值设置），训练预测评估模型，获取最佳模型；（5分）
        MultiClassClassificationTuningEvaluator multiClassClassificationTuningEvaluator = new MultiClassClassificationTuningEvaluator()
                .setPredictionDetailCol("pred_detail")
                .setPredictionCol("pred")
                .setLabelCol(label)
                .setTuningMultiClassMetric(TuningMultiClassMetric.ACCURACY);

        //knn
        ParamGrid paramGrid1 = new ParamGrid()
                .addGrid(knn, KnnClassifier.K, new Integer[] {1, 2, 3})
                .addGrid(knn, KnnClassifier.NUM_THREADS, new Integer[] {3, 6, 9});

        GridSearchCV cv1 = new GridSearchCV()
                .setEstimator(knn)
                .setParamGrid(paramGrid1)
                .setTuningEvaluator(multiClassClassificationTuningEvaluator)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");

        PipelineModel knn_model = cv1.fit(validateData).getBestPipelineModel();
        BatchOperator<?> knn_transform1 = knn_model.transform(testData);
        //rfc
        ParamGrid paramGrid2 = new ParamGrid()
                .addGrid(rfc, RandomForestClassifier.MAX_BINS, new Integer[] {1, 2, 3})
                .addGrid(rfc, RandomForestClassifier.NUM_THREADS, new Integer[] {3, 6, 9});

        GridSearchCV cv2 = new GridSearchCV()
                .setEstimator(rfc)
                .setParamGrid(paramGrid2)
                .setTuningEvaluator(multiClassClassificationTuningEvaluator)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");

        PipelineModel rfc_model = cv2.fit(validateData).getBestPipelineModel();
        BatchOperator<?> rfc_transform1 = rfc_model.transform(testData);
        //（6）、构建分类模型：使用最佳2个分类模型对测试数据集testData进行预测，并计算分类评估准确率Accuracy值；（5分）
        MultiClassMetrics knn_multiClassMetrics1 = metrics.linkFrom(knn_transform1).collectMetrics();
        System.out.println("knn 准确率Accuracy值:"+knn_multiClassMetrics1.getAccuracy());

        MultiClassMetrics rfc_multiClassMetrics1 = metrics.linkFrom(rfc_transform1).collectMetrics();
        System.out.println("rfc 准确率Accuracy值:"+rfc_multiClassMetrics1.getAccuracy());
        //（7）、构建聚类模型：调用Alink中KMeans聚类API，使用葡萄酒数据集构建聚类模型，并且评估模型输出CH轮廓系数值；（4分）
        VectorAssembler res = new VectorAssembler()
                .setSelectedCols(features)
                .setOutputCol("vec");
        BatchOperator<?> transform = res.transform(trainData);

        KMeans kmeans = new KMeans()
                .setVectorCol("vec")
                .setK(2)
                .setPredictionCol("pred");
        BatchOperator<?> transform1 = kmeans.fit(transform).transform(transform);

        ClusterMetrics metrics1 = new EvalClusterBatchOp()
                .setVectorCol("vec")
                .setPredictionCol("pred")
                .linkFrom(transform1)
                .collectMetrics();

        System.out.println(metrics1.getVrc());
    }
}
